package com.atguigu.actual.edu0417.app.dwd.db;

import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.app.SqlBaseApp;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author: techyang
 * @since: 2023-10-20 14:14
 * @description: 交易域下单事务事实表
 **/
public class DwdTradeOrderDetail extends SqlBaseApp {

    public static void main(String[] args) {
        DwdTradeOrderDetail dwdTradeOrderDetail = new DwdTradeOrderDetail();
        dwdTradeOrderDetail.base(args);
    }

    @Override
    public void process(StreamTableEnvironment tableEnv) {

        // 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        // 从topic_db中读取源数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_order_detail_group"));

        // 从topic_db中过滤出订单明细表数据
        Table detailTable = tableEnv.sqlQuery(" select ts, " +
                "data['id'] id, " +
                "data['course_id'] course_id, " +
                "data['course_name'] course_name, " +
                "data['order_id'] order_id, " +
                "data['user_id'] user_id, " +
                "data['origin_amount'] origin_amount, " +
                "data['coupon_reduce'] coupon_reduce, " +
                "data['final_amount'] final_amount " +
                "from topic_db  " +
                "where `table` = 'order_detail'  " +
                "and `type` = 'insert' ");
        tableEnv.createTemporaryView("order_detail",detailTable);
        // tableEnv.executeSql(" select * from order_detail ").print();
        // 从topic_db中过滤出订单信息表数据
        Table orderTable = tableEnv.sqlQuery(" select " +
                "data['id'] id, " +
                "data['order_status'] order_status, " +
                "data['out_trade_no'] out_trade_no, " +
                "data['session_id'] session_id, " +
                "data['province_id'] province_id  " +
                "from topic_db " +
                "where `table` = 'order_info' " +
                "and `type` = 'insert' ");
        tableEnv.createTemporaryView("order_info",orderTable);
        // tableEnv.executeSql(" select * from order_info ").print();

        // 读取页面日志数据
        tableEnv.executeSql(" create table page_log (" +
                " common map<string,string>, " +
                " page map<string,string>, " +
                " ts string " +
                ") " + MyKafkaUtil.getKafkaDDL("dwd_traffic_page_log","dwd_trade_order_detail_group"));
        // tableEnv.executeSql(" select * from page_log ").print();

        // 过滤出order下单结算页面的日志
        Table orderPageTable = tableEnv.sqlQuery(" select page['page_id'] page_id,common['sc'] sc,common['sid'] sid " +
                " from page_log where page is not null and page['page_id'] = 'order' ");
        tableEnv.createTemporaryView("order_page",orderPageTable);
        // tableEnv.executeSql(" select * from order_page ").print();

        // 将三张表进行join
        Table joinTable = tableEnv.sqlQuery("select " +
                "od.id, " +
                "od.course_id, " +
                "od.course_name, " +
                "od.order_id, " +
                "od.user_id, " +
                "od.origin_amount, " +
                "od.coupon_reduce, " +
                "od.final_amount, " +
                "oi.order_status," +
                "oi.out_trade_no," +
                "oi.session_id," +
                "oi.province_id," +
                "op.sc source_id," +
                "od.ts " +
                "from order_info oi join order_detail od " +
                "on oi.id = od.order_id " +
                "join order_page op " +
                "on oi.session_id = op.sid ");
        tableEnv.createTemporaryView("join_table",joinTable);
        // tableEnv.executeSql(" select * from  join_table ").print();
        // 创建写入kafka主题的动态表
        tableEnv.executeSql(" create table dwd_trade_order_detail " +
                " ( " +
                "   id string, " +
                "   course_id string, " +
                "   course_name string, " +
                "   order_id string, " +
                "   user_id string, " +
                "   origin_amount string, " +
                "   coupon_reduce string, " +
                "   final_amount string, " +
                "   order_status string, " +
                "   out_trade_no string, " +
                "   session_id string, " +
                "   province_id string, " +
                "   source_id string, " +
                "   ts string, " +
                "   primary key(id) not enforced " +
                " ) " +
                MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail")
                );
        // 写入kafka
        tableEnv.executeSql(" insert into dwd_trade_order_detail select * from join_table ");
    }
}
